package com.example.rocketmq.service;

import com.example.rocketmq.config.RocketMQProperties;
import com.example.rocketmq.model.Message;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 消息生产者服务
 * 
 * @author demo
 */
@Slf4j
@Service
public class MessageProducerService {

    @Autowired
    private RocketMQProperties properties;

    private DefaultMQProducer producer;
    private final AtomicLong messageCounter = new AtomicLong(0);

    @PostConstruct
    public void init() throws Exception {
        producer = new DefaultMQProducer(properties.getProducer().getGroup());
        producer.setNamesrvAddr(properties.getNamesrvAddr());
        producer.setSendMsgTimeout(properties.getProducer().getSendMessageTimeout());
        producer.setRetryTimesWhenSendAsyncFailed(2);
        producer.start();
        log.info("消息生产者启动成功，NameServer: {}, 生产者组: {}", properties.getNamesrvAddr(), properties.getProducer().getGroup());
    }

    @PreDestroy
    public void destroy() {
        if (producer != null) {
            producer.shutdown();
            log.info("消息生产者已关闭");
        }
    }

    /**
     * 发送同步消息
     */
    public SendResult sendSyncMessage(Message message) throws Exception {
        MessageExt msg = convertToMessageExt(message);
        SendResult result = producer.send(msg);
        log.info("同步发送消息成功，消息ID: {}, 发送结果: {}", message.getMessageId(), result);
        return result;
    }

    /**
     * 发送异步消息
     */
    public CompletableFuture<SendResult> sendAsyncMessage(Message message) {
        CompletableFuture<SendResult> future = new CompletableFuture<>();
        try {
            MessageExt msg = convertToMessageExt(message);
            producer.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    log.info("异步发送消息成功，消息ID: {}, 发送结果: {}", message.getMessageId(), sendResult);
                    future.complete(sendResult);
                }

                @Override
                public void onException(Throwable e) {
                    log.error("异步发送消息失败，消息ID: {}", message.getMessageId(), e);
                    future.complete(null);
                }
            });
        } catch (Exception e) {
            log.error("发送异步消息异常，消息ID: {}", message.getMessageId(), e);
            future.complete(null);
        }
        return future;
    }

    /**
     * 批量发送消息
     */
    public void sendBatchMessages(int count) {
        log.info("开始批量发送 {} 条消息", count);
        for (int i = 0; i < count; i++) {
            Message message = createTestMessage(i + 1);
            try {
                sendAsyncMessage(message);
                Thread.sleep(10); // 短暂延迟，避免发送过快
            } catch (Exception e) {
                log.error("批量发送消息失败，索引: {}", i, e);
            }
        }
        log.info("批量发送消息完成，共发送 {} 条", count);
    }

    /**
     * 创建测试消息
     */
    private Message createTestMessage(int index) {
        return Message.builder()
                .messageId(UUID.randomUUID().toString())
                .content("测试消息-" + index)
                .type("TEST")
                .createTime(LocalDateTime.now())
                .businessData("业务数据-" + index)
                .priority(index % 3 + 1)
                .retryCount(0)
                .build();
    }

    /**
     * 转换消息格式
     */
    private MessageExt convertToMessageExt(Message message) {
        MessageExt msg = new MessageExt();
        msg.setTopic(properties.getTopic().getName());
        msg.setBody(message.toString().getBytes());
        msg.setKeys(message.getMessageId());
        msg.setTags(message.getType());
        return msg;
    }

    /**
     * 获取消息计数器
     */
    public long getMessageCount() {
        return messageCounter.get();
    }
}
